package mr

import (
	"encoding/json"
	"fmt"
	"io/ioutil"
	"os"
	"path/filepath"
	"sort"
	"strconv"
	"time"
)
import "log"
import "net/rpc"
import "hash/fnv"

//
// Map functions return a slice of KeyValue.
//

type KeyValue struct {
	Key   string
	Value string
}
type ByKey []KeyValue

func (a ByKey) Len() int           { return len(a) }
func (a ByKey) Swap(i, j int)      { a[i], a[j] = a[j], a[i] }
func (a ByKey) Less(i, j int) bool { return a[i].Key < a[j].Key }

type Response struct {
	Id           int //任务id
	State        int
	Input        string   //输入文件
	Intermediate []string //中间文件位置
	Output       string   //reduce任务输出结果
	NReduce      int
	NFiles       int
}

//
// use ihash(key) % NReduce to choose the reduce
// task number for each KeyValue emitted by Map.
//
func ihash(key string) int {
	h := fnv.New32a()
	h.Write([]byte(key))
	return int(h.Sum32() & 0x7fffffff)
}

//
// main/mrworker.go calls this function.
//
func Worker(mapf func(string, string) []KeyValue,
	reducef func(string, []string) string) {
	for {
		res := getTask()
		switch res.State {
		case MAP:
			doMapTask(mapf, &res)
			break
		case REDUCE:
			doReduceTask(reducef, &res)
			break
		case WAIT:
			time.Sleep(10 * time.Second)
			break
		case EXIT:
			break
		default:
			return
		}
	}
}

/*获取任务*/
func getTask() Response {
	args := Response{}
	response := Response{}
	call("Coordinator.Assign", args, &response)
	return response
}

/*执行map*/
func doMapTask(mapf func(string, string) []KeyValue, w *Response) {
	//打开文件
	intermediate := []KeyValue{}
	infile, err := os.Open(w.Input)
	if err != nil {
		log.Fatalf("cannot open %v", w.Input)
	}
	content, err := ioutil.ReadAll(infile)
	if err != nil {
		log.Fatalf("cannot read %v", w.Input)
	}
	infile.Close()
	//执行map函数
	intermediate = mapf(w.Input, string(content))
	//保存在nReduce个文件中
	files := make([]*os.File, w.NReduce)
	filesName := make([]string, w.NReduce)
	array := [][]KeyValue{}
	for i := 0; i < w.NReduce; i++ {
		array = append(array, make(ByKey, 0))
	}
	//保存在nReduce个数组中
	for i := 0; i < len(intermediate); i++ {
		ele := intermediate[i]
		index := ihash(ele.Key) % w.NReduce
		array[index] = append(array[index], ele)
	}
	//创建并保存在nReduce个文件中
	for i := 0; i < w.NReduce; i++ {
		//oname :=
		//ofile, err := os.Create(oname)
		files[i], _ = ioutil.TempFile("mr-tmp", "mr-tmp-*")
		enc := json.NewEncoder(files[i])
		enc.Encode(&array[i])
		if err != nil {
			return
		}
	}
	for i, file := range files {
		newName := "mr-" + strconv.Itoa(w.Id) + "-" + strconv.Itoa(i) + ".json"
		path := filepath.Join(file.Name())
		os.Rename(path, newName)
		file.Close()
	}
	w.Intermediate = filesName
	reply := Response{}
	//发送报告
	call("Coordinator.HandleReport", &w, reply)
}

/*执行reduce*/
func doReduceTask(reducef func(string, []string) string, res *Response) {
	intermediate := []KeyValue{}
	//读取中间键并排序
	for i := 0; i < res.NFiles; i++ {
		oname := "mr-" + strconv.Itoa(i) + "-" + strconv.Itoa(res.Id) + ".json"
		file, err := os.Open(oname)
		if err != nil {
			log.Fatalf("cannot read %v", res.Intermediate[i])
		}
		dec := json.NewDecoder(file)
		var data []KeyValue
		err = dec.Decode(&data)
		if err != nil {
			break
		}
		intermediate = append(intermediate, data...)
	}
	sort.Sort(ByKey(intermediate))
	//对相同的key执行reducef
	//oname := "mr-out-" + strconv.Itoa(res.Id) + ".txt"
	//ofile, _ := os.Create(oname)
	ofile, _ := ioutil.TempFile("mr-tmp", "mr-tmp-*")
	i := 0
	//对取到的所有kv进行整理，得到key,list(value)
	for i < len(intermediate) {
		j := i + 1
		for j < len(intermediate) && intermediate[j].Key == intermediate[i].Key {
			j++
		}
		values := []string{}
		for k := i; k < j; k++ {
			values = append(values, intermediate[k].Value)
		}
		output := reducef(intermediate[i].Key, values)

		// this is the correct format for each line of Reduce Output.
		fmt.Fprintf(ofile, "%v %v\n", intermediate[i].Key, output)
		i = j
	}
	os.Rename(filepath.Join(ofile.Name()), "mr-out-"+strconv.Itoa(res.Id)+".txt")
	reply := Response{}
	//发送报告
	call("Coordinator.HandleReport", &res, reply)
}

/*rpc*/
func call(rpcname string, args interface{}, reply interface{}) bool {
	c, err := rpc.DialHTTP("tcp", "127.0.0.1:8080")
	if err != nil {
		log.Fatal("dialing:", err)
	}
	defer c.Close()

	err = c.Call(rpcname, args, reply)
	if err == nil {
		return true
	}

	return false
}
